package com.aotain.nyx.flink;

import java.io.FileInputStream;
import java.io.InputStream;
import java.util.Properties;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;

import com.aotain.common.ZkKafkaConfig;
import com.aotain.nyx.abnormal.AbnormalCheckImport;
import com.aotain.nyx.abnormal.AbnormalReduce;
import com.aotain.nyx.abnormal.AbnormalStatMap;
import com.aotain.nyx.abnormal.AbnormalStatTuple;
import com.aotain.nyx.ackflood.AckFloodFlink;
import com.aotain.nyx.ackflood.AckFloodReduce;
import com.aotain.nyx.ackflood.AckFloodStatMap;
import com.aotain.nyx.ackflood.AckFloodTuple;
import com.aotain.nyx.allattstatd.AllAttDStatisImport;
import com.aotain.nyx.allattstatd.AllAttStatMap;
import com.aotain.nyx.allattstatd.AllAttStatisReduce;
import com.aotain.nyx.appstatd.AppAttDStatisImport;
import com.aotain.nyx.appstatd.AppAttStatMap;
import com.aotain.nyx.appstatd.AppAttStatisReduce;
import com.aotain.nyx.apptype.AppTypeAllStatMap;
import com.aotain.nyx.apptype.AppTypeAllStatisReduce;
import com.aotain.nyx.apptype.AppTypeDAllStatisImport;
import com.aotain.nyx.apptype.AppTypeDStatisImport;
import com.aotain.nyx.apptype.AppTypeStatMap;
import com.aotain.nyx.apptype.AppTypeStatisReduce;
import com.aotain.nyx.bots.BotsFlink;
import com.aotain.nyx.bots.BotsReduce;
import com.aotain.nyx.bots.BotsStatMap;
import com.aotain.nyx.bots.BotsTuple;
import com.aotain.nyx.cc.CCDetectFlink;
import com.aotain.nyx.cc.URLStatMap;
import com.aotain.nyx.cc.URLStatReduce;
import com.aotain.nyx.common.PostTuple;
import com.aotain.nyx.common.URLStatTuple;
import com.aotain.nyx.flowstatd.FlowAttDStatisImport;
import com.aotain.nyx.flowstatd.FlowAttStatMap;
import com.aotain.nyx.flowstatd.FlowAttStatisReduce;
import com.aotain.nyx.ipstat.IPStatTuple;
import com.aotain.nyx.netipstat.AbnStatisMap;
import com.aotain.nyx.netipstat.IPFlowStatImport;
import com.aotain.nyx.netipstat.IPFlowStatMap;
import com.aotain.nyx.netipstat.IPFlowStatReduce;
import com.aotain.nyx.netipstat.IPFlowStatTotalImport;
import com.aotain.nyx.netipstat.IPFlowStatTotalMap;
import com.aotain.nyx.netipstat.IPFlowStatTotalReduce;
import com.aotain.nyx.port.PortDetect;
import com.aotain.nyx.port.PortMap;
import com.aotain.nyx.port.PortReduce;
import com.aotain.nyx.scanattack.ScanAttackFlink;
import com.aotain.nyx.scanattack.ScanAttackReduce;
import com.aotain.nyx.scanattack.ScanAttackStatMap;
import com.aotain.nyx.scanattack.ScanAttackTuple;
import com.aotain.nyx.sensitiveurl.SensitiveUrlDetect;
import com.aotain.nyx.sqlinjection.SQLInjectionDetect;
import com.aotain.nyx.sqlinjection.SQLInjectionMap;
import com.aotain.nyx.sqlinjection.SQLInjectionReduce;
import com.aotain.nyx.sqlinjection.SQLInjectionTuple;
import com.aotain.nyx.statis.AbnStatisTuple;
import com.aotain.nyx.threateva.ThreatenEvaluateImport;
import com.aotain.nyx.threateva.ThreatenEvaluateMap;
import com.aotain.nyx.threateva.ThreatenEvaluateReduce;
import com.aotain.nyx.util.CommonUtil;
import com.aotain.nyx.util.FilterNullEnum;
import com.aotain.nyx.util.FilterNullObject;
import com.aotain.nyx.vc.VCDetect;
import com.aotain.nyx.vc.VCMap;
import com.aotain.nyx.vc.VCReduce;
import com.aotain.nyx.xss.XSSDetect;
import com.aotain.nyx.xss.XSSMap;
import com.aotain.nyx.xss.XSSReduce;
import com.aotain.nyx.xss.XSSTuple;

/**
 * Flink Jupiter 项目主类
 * 网络威胁平台
 * Start Cmd：
 * /usr/local/flink/bin/flink run -c com.aotain.nyx.flink.JupiterFlinkStart 
 * 		/home/hadoop/nyx/lib/hades-nyx-0.0.1-SNAPSHOT.jar
 * @author Administrator
 *
 */
public class MetisFlinkStart {

	public static void main(String[] args) throws Exception {

//		String dbJsonSDS = "{\"driverClassName\":\"com.mysql.jdbc.Driver\","
//				+ "\"maxActive\":\"100\",\"maxIdle\":\"25\",\"maxWait\":\"10000\","
//				+ "\"name\":\"sds\",\"password\":\"h16aug8v3w\",\"queryTimeout\":\"60\","
//				+ "\"type\":\"javax.sql.DataSource\",\"url\":\"jdbc:mysql://192.168.5.28:3306/SDS\""
//				+ ",\"userName\":\"sds\""
//				+ ",\"validationQuery\":\"SELECT SYSDATE()\"}";
//		String dbJsonNYX = "{\"driverClassName\":\"com.mysql.jdbc.Driver\","
//				+ "\"maxActive\":\"100\",\"maxIdle\":\"25\",\"maxWait\":\"10000\","
//				+ "\"name\":\"sds\",\"password\":\"h16aug8v3w\",\"queryTimeout\":\"60\","
//				+ "\"type\":\"javax.sql.DataSource\",\"url\":\"jdbc:mysql://192.168.5.28:3306/NYX\","
//				+ "\"userName\":\"sds\",\"validationQuery\":\"SELECT SYSDATE()\"}";
		Properties prop = new Properties();
		InputStream in = new FileInputStream("/home/storm/config/dbconfig.ini");
		prop.load(in);
		String dbJsonSDS = prop.getProperty("dbJsonSDS");
		String dbJsonNYX = prop.getProperty("dbJsonNYX");
		String dbJsonMETIS = prop.getProperty("dbJsonMETIS");
		
		String filterIpSql = "SELECT DISTINCT A.IP FROM SDS_SERVERINFO A JOIN SDS_WEBSITEINFO B ON A.SITEID = B.SITEID WHERE B.MONITORSTATUS = 1";
		String whiteIpSql = " SELECT * FROM SDS_WHITELIST";
		String appTypeSql = "select APPID,APPTYPE,APPNAME,APPTYPENAME from VIO_CFG_APPTYPE";
		
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


		Properties properties = new Properties();
		properties.setProperty("bootstrap.servers", CommonUtil.KAFKA_BROKER);
		// only required for Kafka 0.8
		properties.setProperty("zookeeper.connect", CommonUtil.ZOOKEEPER_HOST);
		properties.setProperty("group.id", CommonUtil.GROUP_ID_METIS);
//		properties.setProperty("group.id", "flink");
		
		Properties propertiesPost = new Properties();
		propertiesPost.setProperty("bootstrap.servers", CommonUtil.KAFKA_BROKER);
		// only required for Kafka 0.8
		propertiesPost.setProperty("zookeeper.connect", CommonUtil.ZOOKEEPER_HOST);
		propertiesPost.setProperty("group.id", CommonUtil.GROUP_ID_POST);
		
		ZkKafkaConfig z_kConfig = new ZkKafkaConfig("/home/storm/config/dbconfig.ini");
		//ZkKafkaConfig z_kConfig = new ZkKafkaConfig("D:\\workspace\\Hades-Z\\hades-main\\hades-project\\config\\dbconfig.ini");

		DataStream<Tuple2<String,String>> sourceStream = null;
		
		//分流输出
		if(z_kConfig.getFlinkMode().toUpperCase().equals("KAFKA")) {
			sourceStream = env.addSource(new FlinkKafkaConsumer08<>(CommonUtil.TOPIC, new SimpleStringSchema(), properties))
					.map(new SourceDataFilterMap(filterIpSql,whiteIpSql, dbJsonSDS, dbJsonNYX))
					.filter(new FilterNullObject<String>(FilterNullEnum.F1ISNOTNULL)).name("UDII-SOURCE");
		} else if (z_kConfig.getFlinkMode().toUpperCase().equals("SOCKET")) {
			//多个端口监听
			String port = z_kConfig.getFlinkSocketPort();
			for(String p : port.split("\\|")) {
				if(sourceStream == null) {
					sourceStream = env.socketTextStream(z_kConfig.getFlinkSocketIP(), Integer.parseInt(p))
							.map(new SourceDataFilterMap(filterIpSql,whiteIpSql, dbJsonSDS, dbJsonNYX))
							.filter(new FilterNullObject<String>(FilterNullEnum.F1ISNOTNULL))
							.name("UDII-SOURCE-" + p);
					
				} else {
					sourceStream = sourceStream.union(env.socketTextStream(z_kConfig.getFlinkSocketIP(), Integer.parseInt(p))
							.map(new SourceDataFilterMap(filterIpSql,whiteIpSql, dbJsonSDS, dbJsonNYX))
							.filter(new FilterNullObject<String>(FilterNullEnum.F1ISNOTNULL))
							.name("UDII-SOURCE-" + p));
				}
			}
			
			if(sourceStream == null) {
				System.out.println("sourceStream ERROR");
				return;
			}
		} else {
			System.out.println("Config Flink Mode is ERROR");
			return;
		}

		
		/*2017-07-25 新增Post 数据接入 by turk*/
		DataStream<Tuple2<String,PostTuple>> sourceStreamPost = env.addSource(new FlinkKafkaConsumer08<>(CommonUtil.TOPIC_POST, new SimpleStringSchema(), propertiesPost))
				.map(new SourceDataPost(filterIpSql,whiteIpSql, dbJsonSDS, dbJsonNYX))
				.filter(new FilterNullObject<PostTuple>(FilterNullEnum.F1ISNOTNULL)).name("POST-SOURCE");
		
		SplitStream<Tuple2<String,String>> splitStream = sourceStream.split(new SourceDataSelector());
		DataStream<Tuple2<String, String>> dipStream = splitStream.select("MetisDipStream")
				.map(new DipSoucreStreamStat());
		
		DataStream<Tuple2<String, String>> sipStream = splitStream.select("MetisSipStream")
				.map(new SipSoucreStreamStat());
		
		//** 应用分类统计METIS_APP_IP_D
		sipStream.map(new AppTypeStatMap())
				.keyBy(0)
				.timeWindow(Time.minutes(5))
				.reduce(new AppTypeStatisReduce())
				.flatMap(new AppTypeDStatisImport(appTypeSql,dbJsonNYX));
		
		//** 应用分类统计METIS_APP_STAT_IP_D
		sipStream.map(new AppTypeAllStatMap())
				.keyBy(0)
				.timeWindow(Time.minutes(5))
				.reduce(new AppTypeAllStatisReduce())
				.flatMap(new AppTypeDAllStatisImport(appTypeSql,dbJsonNYX)).name("METIS_APP_STAT_IP_D");
			
		//** IP 流量统计
		DataStream<Tuple2<String, IPStatTuple>> destIPStatStream = dipStream.map(new IPFlowStatMap())
					.name("DESTIP-FlowStatMap-MAP"); 
		
		DataStream<Tuple2<String, IPStatTuple>> sourceIPStatStream = sipStream.map(new IPFlowStatMap())
					.name("SOURCEIP-FlowStatMap-MAP"); 

		/*统计 主机 终端的 上下行流量*/
		SingleOutputStreamOperator<IPStatTuple>  
			time5Min = destIPStatStream.union(sourceIPStatStream).keyBy(0)
							       .timeWindow(Time.minutes(3))
							       .reduce(new IPFlowStatReduce())
							       .flatMap(new IPFlowStatImport("METIS_FLOW_STAT_5MIN")).name("IPSTAT-5MIN");
		
		time5Min.map(new IPFlowStatTotalMap())
				.keyBy(0)
				.timeWindow(Time.minutes(3))
				.reduce(new IPFlowStatTotalReduce())
				.flatMap(new IPFlowStatTotalImport("METIS_ALL_FLOW_STAT_5MIN")).name("TOTAL STAT 5MIN");

		SingleOutputStreamOperator<IPStatTuple>  
			time30Min = destIPStatStream.union(sourceIPStatStream).keyBy(0)
				                    .timeWindow(Time.minutes(5))
									.reduce(new IPFlowStatReduce())
									.flatMap(new IPFlowStatImport("METIS_FLOW_STAT_30MIN")).name("IPSTAT-30MIN");

		time30Min.map(new IPFlowStatTotalMap())
			.keyBy(0)
			.timeWindow(Time.minutes(5))
			.reduce(new IPFlowStatTotalReduce())
			.flatMap(new IPFlowStatTotalImport("METIS_ALL_FLOW_STAT_30MIN")).name("TOTAL STAT 30MIN");
		
		SingleOutputStreamOperator<IPStatTuple>  
			time60Min = destIPStatStream.union(sourceIPStatStream).keyBy(0)
				                    .timeWindow(Time.minutes(5))
				                    .reduce(new IPFlowStatReduce())
				                    .flatMap(new IPFlowStatImport("METIS_FLOW_STAT_HOUR")).name("IPSTAT-HOUR");
		
		time60Min.map(new IPFlowStatTotalMap())
			.keyBy(0)
			.timeWindow(Time.minutes(5))
			.reduce(new IPFlowStatTotalReduce())
			.flatMap(new IPFlowStatTotalImport("METIS_ALL_FLOW_STAT_HOUR")).name("TOTAL STAT HOUR");
				
		SingleOutputStreamOperator<IPStatTuple>
			timeDay = destIPStatStream.union(sourceIPStatStream).keyBy(0)
								  .timeWindow(Time.minutes(5))
								  .reduce(new IPFlowStatReduce())
								  .flatMap(new IPFlowStatImport("METIS_FLOW_STAT_DAY")).name("IPSTAT-DAY");
		
		timeDay.map(new IPFlowStatTotalMap())
			.keyBy(0)
			.timeWindow(Time.minutes(5))
			.reduce(new IPFlowStatTotalReduce())
			.flatMap(new IPFlowStatTotalImport("METIS_ALL_FLOW_STAT_DAY")).name("TOTAL STAT DAY");
		
		
		//**端口异常检测
		SingleOutputStreamOperator<AbnStatisTuple>  
			winPort =sipStream.union(dipStream).map(new PortMap(dbJsonSDS))
						       .filter(new FilterNullObject<BasicTuple>(FilterNullEnum.F1ISNOTNULL))
				               .keyBy(0).timeWindow(Time.minutes(1))
						       .reduce(new PortReduce())
						       .flatMap(new PortDetect(dbJsonSDS)).name("Port Detect");
		
		//**CC 攻击检测
		SingleOutputStreamOperator<AbnStatisTuple>  
			winCC = sipStream.union(dipStream).map(new URLStatMap())
			           .filter(new FilterNullObject<URLStatTuple>(FilterNullEnum.F1ISNOTNULL))
		               .keyBy(0)
		               .timeWindow(Time.minutes(1))
		               .reduce(new URLStatReduce())
		               .flatMap(new CCDetectFlink()).name("CC");
		//sql注入
		SingleOutputStreamOperator<AbnStatisTuple>  
		 	winSQLInjection = sipStream.union(dipStream).map(new SQLInjectionMap(z_kConfig.getSQLInjection()))
				       .filter(new FilterNullObject<SQLInjectionTuple>(FilterNullEnum.F1ISNOTNULL))
		               .keyBy(0)
		               .timeWindow(Time.minutes(1))
		               .reduce(new SQLInjectionReduce())
		               .flatMap(new SQLInjectionDetect()).name("SQL Injection");
		
		 /*Sensitive URL*/
	    SingleOutputStreamOperator<AbnStatisTuple>  
			winSensitiveUrl = sipStream.union(dipStream).map(new URLStatMap())
			           .filter(new FilterNullObject<URLStatTuple>(FilterNullEnum.F1ISNOTNULL))
		               .keyBy(0)
		               .flatMap(new SensitiveUrlDetect()).name("Sensitive Url");
		
		SingleOutputStreamOperator<AbnStatisTuple>  
			winXSS = sipStream.union(dipStream).map(new XSSMap())
				       .filter(new FilterNullObject<XSSTuple>(FilterNullEnum.F1ISNOTNULL))
		               .keyBy(0)
		               .timeWindow(Time.minutes(1))
		               .reduce(new XSSReduce())
		               .flatMap(new XSSDetect()).name("XSS Detect");
		
		SingleOutputStreamOperator<AbnStatisTuple>  
			winVC = sipStream.union(dipStream).map(new VCMap())
				       .filter(new FilterNullObject<BasicTuple>(FilterNullEnum.F1ISNOTNULL))
		               .keyBy(0)
		               .timeWindow(Time.minutes(1))
		               .reduce(new VCReduce())
		               .flatMap(new VCDetect()).name("VC Detect");
		
		
		//---异常流量
		SingleOutputStreamOperator<AbnStatisTuple>
			winAbn = sipStream.union(dipStream).map(new AbnormalStatMap())
				   .filter(new FilterNullObject<AbnormalStatTuple>(FilterNullEnum.F1ISNOTNULL))
	               .keyBy(0)
	               .timeWindow(Time.minutes(1))
	               .reduce(new AbnormalReduce())
	               .flatMap(new AbnormalCheckImport()).name("AbnormalCheck");
		
		
		//---僵尸主机
		SingleOutputStreamOperator<AbnStatisTuple>
			winBots = sipStream.union(dipStream).map(new BotsStatMap())
				   .filter(new FilterNullObject<BotsTuple>(FilterNullEnum.F1ISNOTNULL))
	               .keyBy(0)
	               .timeWindow(Time.minutes(1))
	               .reduce(new BotsReduce())
	               .flatMap(new BotsFlink()).name("Bots");
		
		//---扫描攻击
		SingleOutputStreamOperator<AbnStatisTuple>
			winScan = sipStream.union(dipStream).map(new ScanAttackStatMap())
				   .filter(new FilterNullObject<ScanAttackTuple>(FilterNullEnum.F1ISNOTNULL))
	               .keyBy(0)
	               .timeWindow(Time.minutes(1))
	               .reduce(new ScanAttackReduce())
	               .flatMap(new ScanAttackFlink()).name("Scan Attack");
		
		//---AckFlood
		SingleOutputStreamOperator<AbnStatisTuple>
			winAck = sipStream.union(dipStream).map(new AckFloodStatMap())
				   .filter(new FilterNullObject<AckFloodTuple>(FilterNullEnum.F1ISNOTNULL))
	               .keyBy(0)
	               .timeWindow(Time.minutes(1))
	               .reduce(new AckFloodReduce())
	               .flatMap(new AckFloodFlink()).name("Ack Flood");

		
		winCC.union(winXSS).union(winPort)
			.union(winSQLInjection)
			.union(winVC)
			.union(winScan)
			.union(winAck)
			.union(winBots)
			.union(winSensitiveUrl)
			.union(winAbn).map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(3))
			.reduce(new IPFlowStatReduce()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_5MIN"));	
		
		winCC.union(winXSS).union(winPort)
			.union(winSQLInjection)
			.union(winVC)
			.union(winScan)
			.union(winAck)
			.union(winBots)
			.union(winSensitiveUrl)
			.union(winAbn).map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(5))
			.reduce(new IPFlowStatReduce()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_30MIN"));	
		
		winCC.union(winXSS).union(winPort)
			.union(winSQLInjection)
			.union(winVC)
			.union(winScan)
			.union(winAck)
			.union(winBots)
			.union(winSensitiveUrl)
			.union(winAbn).map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(5))
			.reduce(new IPFlowStatReduce()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_HOUR"));	
		
		winCC.union(winXSS).union(winPort)
			.union(winSQLInjection)
			.union(winVC)
			.union(winScan)
			.union(winAck)
			.union(winBots)
			.union(winSensitiveUrl)
			.union(winAbn).map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(5))
			.reduce(new IPFlowStatReduce()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_DAY"));		
				
		//全网 统计
		winCC.union(winXSS).union(winPort)
			.union(winSQLInjection)
			.union(winVC)
			.union(winScan)
			.union(winAck)
			.union(winBots)
			.union(winSensitiveUrl)
			.union(winAbn).map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(5))
			.reduce(new IPFlowStatTotalReduce())
			.flatMap(new IPFlowStatTotalImport("METIS_ALL_FLOW_STAT_5MIN")).name("TOTAL STAT 5MIN");
		
		winCC.union(winXSS).union(winPort)
			.union(winSQLInjection)
			.union(winVC)
			.union(winScan)
			.union(winAck)
			.union(winBots)
			.union(winSensitiveUrl)
			.union(winAbn).map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(5))
			.reduce(new IPFlowStatTotalReduce())
			.flatMap(new IPFlowStatTotalImport("METIS_ALL_FLOW_STAT_30MIN")).name("TOTAL STAT 30MIN");
		winCC.union(winXSS).union(winPort)
			.union(winSQLInjection)
			.union(winVC)
			.union(winScan)
			.union(winAck)
			.union(winBots)
			.union(winSensitiveUrl)
			.union(winAbn).map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(5))
			.reduce(new IPFlowStatTotalReduce())
			.flatMap(new IPFlowStatTotalImport("METIS_ALL_FLOW_STAT_HOUR")).name("TOTAL STAT HOUR");
		//全网 统计(DAY)
		winCC.union(winXSS).union(winPort)
			.union(winSQLInjection)
			.union(winVC)
			.union(winScan)
			.union(winAck)
			.union(winBots)
			.union(winAbn).map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(5))
			.reduce(new IPFlowStatTotalReduce())
			.flatMap(new IPFlowStatTotalImport("METIS_ALL_FLOW_STAT_DAY")).name("TOTAL STAT DAY");
		
		//应用攻击统计天表SDS_APP_ATT_D
		winCC.union(winSQLInjection).union(winPort)
			.union(winXSS)
			.union(winVC)
			.union(winScan)
			.union(winAck)
			.union(winBots)
			.union(winSensitiveUrl)
			.union(winAbn)
			.map(new AppAttStatMap()).keyBy(0).timeWindow(Time.minutes(5))
			.reduce(new AppAttStatisReduce()).flatMap(new AppAttDStatisImport()).name("SDS_APP_ATT_D");
		
		//流量攻击统计天表SDS_FLOW_ATT_D
		winCC.union(winSQLInjection).union(winPort)
			.union(winXSS)
			.union(winVC)
			.union(winScan)
			.union(winAck)
			.union(winBots)
			.union(winSensitiveUrl)
			.union(winAbn)
			.map(new FlowAttStatMap()).keyBy(0).timeWindow(Time.minutes(5))
			.reduce(new FlowAttStatisReduce()).flatMap(new FlowAttDStatisImport()).name("SDS_FLOW_ATT_D");
		
		//攻击全量统计天表SDS_ALL_ATT_D
		winCC.union(winSQLInjection).union(winPort)
		.union(winXSS)
		.union(winVC)
		.union(winScan)
		.union(winAck)
		.union(winBots)
		.union(winSensitiveUrl)
		.union(winAbn)
		.map(new AllAttStatMap()).keyBy(0).timeWindow(Time.minutes(5))
		.reduce(new AllAttStatisReduce()).flatMap(new AllAttDStatisImport()).name("SDS_ALL_ATT_D");
		
		//评分表METIS_THREATEN_EVALUATE
		winCC.union(winSQLInjection).union(winPort)
		.union(winXSS)
		.union(winVC)
		.union(winScan)
		.union(winAck)
		.union(winBots)
		.union(winSensitiveUrl)
		.union(winAbn)
		.map(new ThreatenEvaluateMap()).keyBy(0).timeWindow(Time.minutes(5))
		.reduce(new ThreatenEvaluateReduce()).flatMap(new ThreatenEvaluateImport(dbJsonMETIS)).name("METIS_THREATEN_EVALUATE");
		
		env.getConfig().setTaskCancellationInterval(60000);
		System.out.println("Cancel Time:" + env.getConfig().getTaskCancellationInterval());
		System.out.println("************************");
		System.out.println("*****planVisualizer*****");
		System.out.println("************************");
		System.out.println(env.getExecutionPlan());
		env.execute("Flink-Jupiter");
	}
}

    